package com.baishancloud.log.douyin.demand2.subtask

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.douyin._
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 由于头条  域名数据从金华地区切换到石家庄地区 kafka 集群，因此抖音的请求需要两个地方进行写入，故在原来的基础上，在 tag 标签中增加 kafka_source 字段来群是消费的哪个地区的 kafka 集群数据。<br>
 * 该任务消费石家庄地区 kafka 数据，运行在石家庄地区 yarn 集群上。
 *
 * @author ziqiang.wang 
 * @date 2022/3/2 14:08
 */
object ColdRatioSjz extends Serializable {

  def main(args: Array[String]): Unit = {

    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println("输入参数：" + parameterTool.toMap.toString)
    if (!paramCheck(parameterTool)) {
      return
    }
    StreamEnv.builder(parameterTool).build()

    val cache1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, "-cache-石家庄").uid("9abe44c8-e53e-40e1-890f-7e1bc48725bf")

    cache1
      .rebalance
      .map(Origin.parse(_).orNull).name("Origin").uid("ccb38504-1b92-446b-9cab-5d04e4a1b365")
      .filter(new DataFilterSjz(parameterTool)).name("DataFilter").uid("f2c6d262-798d-4dd1-8383-49b80341eb4a")
      .process(new LogParse2Sub(kafkaSourceSjz)).name("logParse").uid("8148b72-1252-4f49-a149-f46e1e714dbb")
      .process(new LocalAgg2Sub(parameterTool)).name("localAgg").uid("d958a5b0-bf5b-404a-9a62-8d4c517de458")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowSize, 1))))
      .reduce((x, y) => Result2Sub(x.time, x.step, x.name, x.endpoint, x.value + y.value, x.tags)).name("windowAgg").uid("2511df02-56c6-4615-a9e3-8c77271d7f85")
      .keyBy(_.getKey)
      .process(new AggAndOut2Sub(parameterTool)).name("aggAndOut").uid("72ced093-87ea-40e2-8fac-31a992d13b9e")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("29b705b6-6908-4d18-92ef-d90a358fdb94")
      .addSink(SinkUtil.mallard(parameterTool.getInt(outThreshold, Int.MaxValue), parameterTool.getLong(outInterval, 5))).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("douyin_cold_ratio").uid("bae08477-0167-4b13-af50-df5f9e086c56")

    StreamEnv.execute(parameterTool)
  }


}
